/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.wb.picmover.util;

import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.session.receive.REAbstractReceiveMessageHandler;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSessionBuilder;
import java.util.Map;

/**
 *
 * @author zwlin
 */
public class RocketMqReceiver {

    static int i = 0;

    public static void main(String[] args) throws REConnectionException {
//        REConnection conn = new REConnection("192.168.11.94:9879");
        REConnection conn = new REConnection("10.136.64.28:9877");
//        jiyutest_mq_2  MyGroupNamejiyu5656
//        REReceiveSessionBuilder builder = (REReceiveSessionBuilder) conn.getReceiveSessionBuilder("wx_gmsg_pic_mov_mg");
        REReceiveSessionBuilder builder = (REReceiveSessionBuilder) conn.getReceiveSessionBuilder("twx_gmsg_update_mq");//swbpic_mq5656
//         REReceiveSessionBuilder builder = (REReceiveSessionBuilder) conn.getReceiveSessionBuilder("spftest1_mq");
        builder.setHandler(new REAbstractReceiveMessageHandler<REMessage>() {
            @Override
            public boolean handle(REMessage message) {

                //column name
                System.out.println("=== === === === === === === === === ===");
                System.out.println("column name :");
                for (String name : message.colNames) {
                    System.out.print(name + "\t");
                }
                System.out.println("");
                System.out.println("");

                //column type
                System.out.println("column type :");
                for (REFieldType type : message.types) {
                    System.out.print(type + "\t");
                }
                System.out.println("");
                System.out.println("");

                //doc schema name
                System.out.println("doc_schema_name :");
                System.out.println(message.docSchemaName == null ? "(null)" : message.docSchemaName);
                System.out.println("");

                //user Desc
                System.out.println("user desc :");
                if (message.userDesc == null) {
                    System.out.println("(null)");
                } else {
                    for (Map.Entry<String, String> entry : message.userDesc.entrySet()) {
                        System.out.println(entry.getKey() + "\t:\t" + entry.getValue());
                    }
                }
                System.out.println("");

                //data
                System.out.println("data :");
                i++;
                for (String name : message.colNames) {
                    System.out.print(name + "\t");
                }
                System.out.println("");
                int colLen = message.colNames.length;
                for (Object[] doc : message.data) {
                    System.out.println("%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%" + i);
                    for (int i = 0; i < colLen; ++i) {
                        if (doc[i] == null) {
                            System.out.print("(null)\t");
                            continue;
                        }
                        switch (message.types[i]) {
                            case Strings:
                                for (String item : (String[]) doc[i]) {
                                    System.out.print(item + ",");
                                }
                                System.out.print("\t");
                                break;
                            case String:
                            default:
                                System.out.print(doc[i].toString() + "\t");
                        }
                    }
                    System.out.println(" ");
                }
//				System.exit(0);
                return true;
            }
        });
        builder.setFailureHandler(new REAbstractReceiveMessageHandler<byte[]>() {
            @Override
            public boolean handle(byte[] message) {
                System.out.println("formatter fuilure ! size = " + message.length);
                return true;
            }
        });
//        builder.setGroupName("MyGroupNamejijiyu3434");
//        builder.setGroupName("swbpic_mq5656");
//        builder.setGroupName("MyGroupNamejijiyu676767");
        builder.setGroupName("iie_groupname33333");

        REReceiveSession session = (REReceiveSession) builder.build();
        session.start();
    }
}
